package com.zlf.sb.demo.service.impl;

import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zlf.sb.demo.consumer.InsertDataConsumer;
import com.zlf.sb.demo.dto.UserEntityDto;
import com.zlf.sb.demo.entity.UserEntity;
import com.zlf.sb.demo.entity.UserJsonDataEntity;
import com.zlf.sb.demo.inquery.BatchParam;
import com.zlf.sb.demo.inquery.SplitQuery;
import com.zlf.sb.demo.mapper.UserJsonDataMapper;
import com.zlf.sb.demo.mapper.UserMapper;
import com.zlf.sb.demo.producer.QueryDoBizProducer;
import com.zlf.sb.demo.service.IUserService;
import com.zlf.sb.demo.service.ThreadPoolService;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.cursor.Cursor;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class UserServiceImpl extends ServiceImpl<UserMapper, UserEntity> implements IUserService {
    @Autowired
    private UserMapper userMapper;

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
    private UserJsonDataMapper userJsonDataMapper;

    @Autowired
    private UserJsonDataServiceImpl userJsonDataService;

    private AtomicInteger queryCount = new AtomicInteger(0);

    private AtomicInteger insertCount = new AtomicInteger(0);

    /**
     * 多线程分页查询
     * 生产者+消费者模型
     * 多线程插入
     */
    @Override
    public void queryInsertData() {
        StopWatch stopWatch = new StopWatch();
        // 开始时间
        stopWatch.start();
        Page<UserEntity> page = new Page<>(1, 10);
        QueryWrapper<UserEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().orderByDesc(UserEntity::getCreateTime);
        Page<UserEntity> pageResult = this.getBaseMapper().selectPage(page, queryWrapper);
        int pageSize = 10000;
        if (Objects.nonNull(pageResult)) {
            long total = pageResult.getTotal();
            int pages = (int) ((total + pageSize - 1) / pageSize);
            log.info("pages:{}", pages);
            ThreadPoolExecutor executor = ThreadPoolService.getInstance();
            CountDownLatch countDownLatch1 = new CountDownLatch(pages);
            CountDownLatch countDownLatch2 = new CountDownLatch(pages);
            LinkedBlockingQueue<List<UserJsonDataEntity>> linkedBlockingQueue = userJsonDataService.getLinkedBlockingQueue();
            for (int i = 1; i <= pages; i++) {
                try {
                    QueryDoBizProducer queryDoBizProducer = new QueryDoBizProducer(linkedBlockingQueue, i, queryCount, countDownLatch1, pageSize);
                    CompletableFuture.runAsync(queryDoBizProducer, executor);
                } catch (Exception e) {
                    log.info("异常1:{}", e.getMessage());
                }
            }
            try {
                countDownLatch1.await();
            } catch (InterruptedException e) {
                log.info("异常2:{}", e.getMessage());
            }
            for (int i = 1; i <= pages; i++) {
                try {
                    InsertDataConsumer insertDataConsumer = new InsertDataConsumer(linkedBlockingQueue, insertCount, countDownLatch2);
                    CompletableFuture.runAsync(insertDataConsumer, executor);
                } catch (Exception e) {
                    log.info("异常3:{}", e.getMessage());
                }
            }
            try {
                countDownLatch2.await();
            } catch (InterruptedException e) {
                log.info("异常4:{}", e.getMessage());
            }
            log.info("queryCount:{}", queryCount);
            log.info("insertCount:{}", insertCount);
        }
        // 结束时间
        stopWatch.stop();
        log.info("执行时长：{}秒", stopWatch.getTotalTimeSeconds());
        // 执行时长：18.903922秒
    }

    /**
     * 游标查询
     */
    @Override
    public void cursorQueryInsertData() {
        StopWatch stopWatch = new StopWatch();
        // 开始时间
        stopWatch.start();
        int total = userMapper.countTotal();
        if (total > 0) {
            int pageSize = 10000;
            int pages = (total + pageSize - 1) / pageSize;
            for (int i = 1; i <= pages; i++) {
                List<UserJsonDataEntity> userJsonDataEntities = new ArrayList<>();
                int start = (i - 1) * pageSize;
                try (SqlSession sqlSession = sqlSessionFactory.openSession(); Cursor<UserEntity> pageCursor = sqlSession.getMapper(UserMapper.class).getCursorPageData(start, pageSize)) {
                    int currentIndex = pageCursor.getCurrentIndex();
                    log.info("currentIndex1:{}", currentIndex);
                    Iterator<UserEntity> iterator = pageCursor.iterator();
                    while (iterator.hasNext()) {
                        UserEntity userEntity = iterator.next();
                        // log.info("userEntity:{}", JSON.toJSONString(userEntity));
                        UserJsonDataEntity jd = com.alibaba.fastjson.JSONObject.parseObject(userEntity.getJsonData(), new TypeReference<UserJsonDataEntity>() {
                        });
                        userJsonDataEntities.add(jd);
                    }
                } catch (Exception e) {
                    log.error("游标分页查询异常:{}", e.getMessage());
                }
                log.info("userJsonDataEntities:{}", userJsonDataEntities.size());
                userJsonDataMapper.insertBatchSomeColumn(userJsonDataEntities);
            }
        }
        // 结束时间
        stopWatch.stop();
        log.info("执行时长：{}秒", stopWatch.getTotalTimeSeconds());
        //执行时长：39.5655331秒
    }

    @Override
    public Page<UserEntity> pageList(UserEntityDto dto) {
        if (Objects.isNull(dto)) {
            throw new RuntimeException("参数不为空!");
        }
        if (Objects.isNull(dto.getCurrent())) {
            dto.setCurrent(1L);
        }
        if (Objects.isNull(dto.getSize())) {
            dto.setSize(100L);
        }
       /* if (CollectionUtil.isNotEmpty(dto.getCredit()) && dto.getCredit().size() == 2) {
            if (dto.getCredit().get(0) > dto.getCredit().get(1)) {
                dto.setCreditStart(dto.getCredit().get(1));
                dto.setCreditEnd(dto.getCredit().get(0));
            } else {
                dto.setCreditStart(dto.getCredit().get(0));
                dto.setCreditEnd(dto.getCredit().get(1));
            }
        }*/
        Page<UserEntity> page = new Page<>(dto.getCurrent(), dto.getSize());
        Long count = this.baseMapper.pageListV2Count(dto);
        if (count == 0) {
            page = new Page<>();
            return page;
        }
        page.setTotal(count);
        Long pages = (count + dto.getSize() - 1) / dto.getSize();
        page.setPages(pages);
        if (page.getCurrent() > pages) {
            page.setCurrent(pages);
        }
        Long pageSize = page.getSize();
        Long start = (page.getCurrent() - 1) * page.getSize();
        if (count < pageSize) {
            pageSize = count;
        }
        Long mod = count % pageSize;
        if (mod != 0 && Objects.equals(dto.getCurrent(), pages)) {
            pageSize = mod;
        }
        //Long end = page.getCurrent() * page.getSize();
        List<UserEntity> memberV2VOList = this.baseMapper.pageListV2(start, pageSize, dto);
        page.setRecords(memberV2VOList);
        return page;
    }

    @Override
    public CompletableFuture<UserEntity> performSecurityCheckAsync(UserEntity user) {
        CompletableFuture<UserEntity> completableFuture  = CompletableFuture.supplyAsync(
                () ->  this.baseMapper.selectById(user.getId()))
                .whenComplete((result, throwable) -> {
                    //触发异常
                    if (throwable != null) {
                        log.error("completableFutureOfSbNumBySblx[whenComplete] error:{}", throwable);
                    }
                });
        return completableFuture;
    }

    @Override
    public CompletableFuture<UserEntity> fetchPreferencesAsync(String userId) {
        CompletableFuture<UserEntity> completableFuture  = CompletableFuture.supplyAsync(
                () ->  this.baseMapper.selectById(userId))
                .whenComplete((result, throwable) -> {
                    //触发异常
                    if (throwable != null) {
                        log.error("completableFutureOfSbNumBySblx[whenComplete] error:{}", throwable);
                    }
                });
        return completableFuture;
    }
    @SplitQuery(batchSize = 2, handlerName = "personResultHandler")
    @Override
    public List<UserEntity> findByAgeAndNameContainingAndIdIn(@BatchParam List<Long> ids) {
        List<UserEntity> userEntityList = this.baseMapper.selectBatchIds(ids);
        return userEntityList;
    }

}
